package my.thesis.bolts.filtersReliable;

import java.util.Map;

import my.thesis.analytics.metrics.ThroughputMetric;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class FilterKeywordReliable extends BaseRichBolt
{
	private static final long serialVersionUID = 1L;
	private OutputCollector collector;
	
	transient ThroughputMetric _throughputMetric;
	private static Boolean metricsOn=false;

	public void execute(Tuple tuple) 
	{
		if(metricsOn)
		{
			_throughputMetric.incrTuples();
		}
		
		Long id = tuple.getLongByField("id");
		String text=tuple.getStringByField("text");
		String username=tuple.getStringByField("username");
		int followers=tuple.getIntegerByField("followers");
		String mention=tuple.getStringByField("mention");
		
		String phrase="big data";
		
		if(text.toLowerCase().contains(phrase)==true)
		{
			//TEST
			//System.out.println("KEYWORD FOUND!");
			//System.out.println(text);
			
			collector.emit(tuple,new Values(id,username,mention,followers));
		}
		
		collector.ack(tuple);
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		declarer.declare(new Fields("id","username","mention","followers"));  		
	}

	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{
		this.collector=collector;
		
		if(metricsOn)
		{
			_throughputMetric=new ThroughputMetric();
			context.registerMetric("filterKeywordCount", _throughputMetric, 20);
			_throughputMetric.setStartTime();
		}
	}

}
